package cn.enjoy.rabbitmq.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


/**
 * @author K
 * @date 2020/11/27 13:24
 */
@Configuration
@Slf4j
public class RabbitMqConfiguration {

    @Bean
    public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {
        // 1开启确认模式 producer发送给exchange时触发
//        connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        // 2开启回退模式 exchange路由到queue失败时触发
        //  A默认 丢弃信息
        //  B返回给消息发送方	需要设置rabbitTemplate.setMandatory(true)参数，将消息退回给producer。并执行回调函数returnedMessage。
        connectionFactory.setPublisherReturns(true);

        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);

        // 1确认模式 定义回调
        // correlationData：相关配置信息
        // ack exchange交换机 true成功 false失败
        // cause 失败原因
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> log.info("消息发送成功:correlationData({}),ack({}),cause({})", correlationData, ack, cause));

        // 2回退模式 定义回调
        // message 消息对象
        // replyCode 错误码
        // replyText 错误信息
        // exchange 交换机
        // routingKey 路由key
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}", exchange, routingKey, replyCode, replyText, message));
//        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
//            @Override
//            public void returnedMessage(ReturnedMessage returned) {
//                log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}", returned.getExchange(), returned.getRoutingKey(), returned.getReplyCode(), returned.getReplyText(), returned.getMessage());
//            }
//        });

        return rabbitTemplate;
    }

    /**
     * 消息消费失败处理
     */
    @Bean
    public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate) {
        return new MqRepublishMessageRecoverer(rabbitTemplate, MqQueueConfig.QUEUE_ERROR, MqQueueConfig.EXCHANGE_ERROR);
    }

}
